package com.mybd.req.two;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import java.io.IOException;

/**
 * @author: Scq
 * @time: 2021/6/510:32
 */
public class JobMain {
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        //1、创建建一个job任务对象
        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration, "mypartitioner");

        //2、指定job所在的jar包
        // job.setJarByClass(PartitionerRunner.class);

        //3、指定源文件的读取方式类和源文件的读取路径
        //按照行读取
        job.setInputFormatClass(TextInputFormat.class);
        // 只需要指定源文件所在的目录即可
        // TextInputFormat.addInputPath(job, new Path("hdfs://node1:8020/input/partitioner"));
        //只需要指定源文件所在的目录即可
        TextInputFormat.addInputPath(job, new Path("file:///C:\\Users\\Scq\\Desktop\\work"));

        //4、指定自定义的Mapper类和K2、V2类型
        //指定Mapper类
        job.setMapperClass(MyMapper.class);
        //K2类型
        job.setMapOutputKeyClass(Text.class);
        //V2类型
        job.setMapOutputValueClass(NullWritable.class);

        //5、指定自定义分区类（如果有的话）
        job.setPartitionerClass(MyPartition.class);
        //6、指定自定义分组类（如果有的话）
        //7、指定自定义的Reducer类和K3、V3的数据类型
        // 指定Reducer类
        // job.setReducerClass(PartitionerReducer.class);
        //K3类型
        job.setOutputKeyClass(Text.class);
        //V3类型
        job.setOutputValueClass(NullWritable.class);


        //设置Reduce的个数
        job.setNumReduceTasks(3);

        //8、指定输出方式类和结果输出路径
        job.setOutputFormatClass(TextOutputFormat.class);
        // 目标目录不能存在，否则报错
        // TextOutputFormat.setOutputPath(job, new  Path("hdfs://node1:8020/output/partitioner"));
        // 目标目录不能存在，否则报错
        TextOutputFormat.setOutputPath(job, new Path("file:///C:\\Users\\Scq\\Desktop\\out2"));
        // job.setGroupingComparatorClass(MyGrouping.class);
        //9、将job提交到yarn集群
        //true表示可以看到任务的执行进度
        boolean bl = job.waitForCompletion(true);

        //10.退出执行进程
        System.exit(bl ? 0 : 1);

    }
}
